// Copyright (c) 2022 Presto Labs Pte. Ltd.
// Author: fengyang

#pragma once
#include <map>
#include <memory>
#include <string>
#include <vector>

#include "coin2/exchange/base/feed/system.h"
#include "coin2/exchange/base/feed/ws/ws_subscriber_util.h"
#include "coin2/exchange/base/feed/ws/subscriber_manager.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/factory/feed/raw_processor.h"
#include "coin2/exchange/base/feed/replay/processor.h"
#include "coin2/flow/topic_subscription.h"

namespace coin2::exchange::di {
class ConstantInjection;
}

namespace coin2::exchange::feed {
class FeedSystemConfig;
}

namespace coin2::exchange::base::executor {
class LiveAsyncExecutor;
}

namespace coin2::exchange::base::symbology {
class IProduct;
}

namespace coin2::exchange::feed {
class FeedSystemConfig;
}

namespace coin2::exchange::base::feed {

class WsSubscriber;

using coin2::exchange::base::feed::replay::ReplayProcessor;
using coin2::exchange::base::feed::counter::FeedCounter;

class LiveParallelFeedSubsystem : public BaseLiveFeedSubsystem {
 public:
  LiveParallelFeedSubsystem(
      const std::string& system_name,
      const MarketExchangeApi& mea,
      const di::ConstantInjection& ci,
      const coin2::exchange::feed::FeedSystemConfig& config,
      coin2::exchange::base::executor::LiveAsyncExecutor* live_executor,
      IFeedSubscriber* callback,
      FeedCounter* counter);

  void StartAsync() override;
  void StopAsync() override;

  std::vector<IProduct*> GetProducts() const override;

  void DebugPrint() const override { LOG(INFO) << "[Live Feed System]"; }

  void GetHealthReport(int indent, std::string* report) const override;

  bool BanProduct(const IProduct* product) override;
  bool BanProduct(const std::vector<IProduct*>& products) override;
  bool BanProduct(const std::vector<const IProduct*>& products) override;

 private:
  executor::TopicRecordSubscriber* NewTopicRecordSubscriber();

  const int32_t num_workers_;
  const bool feed_arbitration_;
  std::shared_ptr<SubscriberManager> subscriber_;
  WsRateLimit rate_limits;
  std::vector<std::unique_ptr<IProduct>> products_;
  TopicRegister topic_register_;
  const coin2::exchange::di::ConstantInjection& ci_;
  const coin2::exchange::feed::FeedSystemConfig& config_;
  IFeedSubscriber* callback_;
  std::map<std::string, executor::TopicRecordSubscriber*> topic2subscriber_;
  std::vector<std::unique_ptr<FeedParsingProcessor>> processors_;
  std::vector<std::unique_ptr<executor::TopicRecordArbitrator>> arbitrators_;
  FeedCounter* feed_counter_;
};

}  // namespace coin2::exchange::base::feed
